package com.ubxtech.mqtt.config;

import cn.hutool.core.util.StrUtil;
import com.ubxtech.mqtt.bo.MqttTopicBO;
import com.ubxtech.mqtt.properties.MqttProperties;
import com.ubxtech.mqtt.service.MqttTopicService;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.inbound.AbstractMqttMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.AbstractMqttMessageHandler;
import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler;
import org.springframework.messaging.MessageChannel;

import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;

/**
 * 可以绑定多个不同的mqtt服务器
 * https://github.com/aLiang-xyl/integration-mqtt
 * https://blog.csdn.net/x2__T/article/details/123138288
 * @author Victor.Xiao
 * @since 2022-11-09 16:08
 **/
@Slf4j
@Configuration
@EnableConfigurationProperties(MqttProperties.class)
@ConditionalOnProperty(prefix = MqttProperties.PREFIX, name = "enabled", havingValue = "v5")
public class Mqttv5MultiConfiguration {

    @Resource
    private MqttProperties mqttProperties;
    @Resource
    private ConfigurableApplicationContext applicationContext;
    @Resource
    private MqttTopicService mqttTopicService;

    @PostConstruct
    public void initChannel(){
        mqttProperties.getConfig().forEach((channelName, config) -> {
            if (config == null || StrUtil.isBlank(config.getClientId()) || config.getUrl() == null
                    || config.getUrl().length == 0 || StrUtil.isBlank(config.getUsername()) || StrUtil.isBlank(config.getPassword())) {
                throw new IllegalArgumentException("mqtt缺少参数配置");
            }
            DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
            beanFactory.registerBeanDefinition(channelName, mqttChannel());
            log.info("注册通道,channel:{},配置:{} ", channelName, config);
            MessageChannel mqttChannel = beanFactory.getBean(channelName, MessageChannel.class);
            String adapterName = channelName + MqttConstant.ADAPTER_NAME_SUFFIX;
            beanFactory.registerBeanDefinition(adapterName, channelAdapter( config, mqttChannel));
            log.info("注册Adapter:{}", adapterName);
            String handlerBeanName = channelName + MqttConstant.OUT_BOUND_CHANNEL_SUFFIX;
            beanFactory.registerBeanDefinition(handlerBeanName, mqttOutbound( config));
            log.info("初始化outBound,{}",handlerBeanName);

            MqttBeanHolderUtil.putMessageHandler(channelName, beanFactory.getBean(handlerBeanName, AbstractMqttMessageHandler.class));
            MqttBeanHolderUtil.putAdapter(channelName, beanFactory.getBean(adapterName, AbstractMqttMessageDrivenChannelAdapter.class));
            //订阅topic
            AbstractMqttMessageDrivenChannelAdapter adapter = MqttBeanHolderUtil.getAdapter(channelName);
            String shareGroup = mqttProperties.getConfig().get(channelName).getShareGroup();
            //服务启动后，将所有设备的主题进行监听
            List<MqttTopicBO> mqttShadowTopics = mqttTopicService.getInitTopics(channelName);
            log.info("订阅topic:{}",mqttShadowTopics);
            Set<String> topics = new HashSet<>();
            mqttShadowTopics.forEach(mqttShadowTopic -> {
                if (!topics.contains(mqttShadowTopic.getSubTopic())) {
                    if (StrUtil.isNotBlank(shareGroup)) {
                        adapter.addTopic("$share/"+shareGroup+"/"+mqttShadowTopic.getSubTopic(), 1);
                    } else {
                        adapter.addTopic(mqttShadowTopic.getSubTopic(), 1);
                    }
                }
                topics.add(mqttShadowTopic.getSubTopic());
            });
        });
    }


    /**
     * 初始化通道
     * @author Victor.Xiao
     * @since 2022-11-09 16:36
     * @return org.springframework.beans.factory.support.AbstractBeanDefinition
    */
    private AbstractBeanDefinition mqttChannel() {
        BeanDefinitionBuilder messageChannelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
        messageChannelBuilder.setScope(BeanDefinition.SCOPE_SINGLETON);
        return messageChannelBuilder.getBeanDefinition();
    }

    /**
     * 配置MqttPahoClientFactory
     * @author Victor.Xiao
     * @since 2022-11-09 16:17
     * @return org.springframework.integration.mqtt.core.MqttPahoClientFactory
    */
    public MqttConnectionOptions mqttConnectionOptions( MqttProperties.Config config) {
        MqttConnectionOptions options = new MqttConnectionOptions();

        options.setServerURIs(config.getUrl());
//        options.setCleanSession(config.getCleanSession() == null ? false : config.getCleanSession());//v3
        options.setCleanStart(config.getCleanStart() != null && config.getCleanStart());
        options.setSessionExpiryInterval(config.getSessionExpiryInterval());
        options.setKeepAliveInterval(config.getKeepAliveInterval() == null ? MqttConstant.KEEP_ALIVE_TIME : config.getKeepAliveInterval());
        options.setPassword(config.getPassword().getBytes());
        options.setUserName(config.getUsername());
        options.setAutomaticReconnect(config.getAutomaticReconnect() == null || config.getAutomaticReconnect());
        options.setConnectionTimeout(config.getTimeout() == null ? MqttConstant.CONNECTION_TIMEOUT : config.getTimeout());
        //v5没有
//        options.setMaxInflight(config.getMaxInflight() == null ? MqttConstant.MAX_IN_FLIGHT : config.getMaxInflight());
        //防止 ERROR o.e.p.c.mqttv3.internal.ClientState - Timed out as no activity  错误
        options.setConnectionTimeout(0);

        return options;
    }

    /**
     * mqtt adapter
     * @author Victor.Xiao
     * @since 2022-11-09 16:43
     * @param config 连接配置
     * @param mqttChannel 通道
     * @return org.springframework.beans.factory.support.AbstractBeanDefinition
    */
    private AbstractBeanDefinition channelAdapter(MqttProperties.Config config, MessageChannel mqttChannel) {
        BeanDefinitionBuilder messageProducerBuilder = BeanDefinitionBuilder.genericBeanDefinition(Mqttv5PahoMessageDrivenChannelAdapter.class);
        messageProducerBuilder.setScope(BeanDefinition.SCOPE_SINGLETON);
        messageProducerBuilder.addConstructorArgValue(mqttConnectionOptions(config));
        messageProducerBuilder.addConstructorArgValue(MqttBeanHolderUtil.appendClientId(config.getClientId() + "_subscribe", config.getClientIdFixed()));
        messageProducerBuilder.addConstructorArgValue(config.getTopics());
//        messageProducerBuilder.addPropertyValue("converter", new DefaultPahoMessageConverter());//v5不需要
        messageProducerBuilder.addPropertyValue("qos", MqttConstant.QOS_1);
        messageProducerBuilder.addPropertyValue("outputChannel", mqttChannel);
        return messageProducerBuilder.getBeanDefinition();
    }

    /**
     * 消息发送客户端
     * @author Victor.Xiao
     * @since 2022-11-09 16:41
     * @param config 连接配置
     * @return org.springframework.beans.factory.support.AbstractBeanDefinition
    */
    private AbstractBeanDefinition mqttOutbound( MqttProperties.Config config) {
        BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(Mqttv5PahoMessageHandler.class);
        builder.addConstructorArgValue(mqttConnectionOptions(config));
        builder.addConstructorArgValue(MqttBeanHolderUtil.appendClientId(config.getClientId() + "_publish", config.getClientIdFixed()));
        //设置异步发送，默认是false(发送时阻塞)
        builder.addPropertyValue("async", config.getAsync() == null || config.getAsync());
        builder.addPropertyValue("defaultQos", MqttConstant.QOS_1);
        builder.addPropertyValue("defaultRetained", false);
        builder.addPropertyValue("asyncEvents", false);

        return builder.getBeanDefinition();
    }
}
